package com.huan.kafka.api;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * 测试kafka消费者
 * 1、消息接收的key和value的序列化
 * 2、指定消费者组
 * 3、自动提交 offset (生产环境可以使用手动提交offset)
 * 4、重置消费者的偏移量，此配置生效的条件
 * 5、自定义消息消费拦截器
 * 6、每次从服务器获取多少数据
 *
 * @author huan.fu 2021/1/5 - 上午10:29
 */
public class KafkaConsumerDemo {
    public static void main(String[] args) throws InterruptedException {
        Properties properties = new Properties();
        // kafka服务器的地址，端口，可以不用全写，写几个既可
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094");
        // kafka 消息key的序列化方式
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        // kafka 消息value的序列化方式
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        // 设置消费者组,多个消费者只要组一样,就认为在一个组中,那个topic中的某一个分区,只能被这个组中的某个消费组消费,消费者组和消费者组之间互不影响
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group-test-001");
        // 自动提交 offset
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        // 自动提交 offset 的间隔，即隔多长时间提交offset
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100);
        // 重置消费者的偏移量，此配置生效的条件：
        //      1、当前消费者的偏移量不在kafka服务器上(比如是一个新的消费者组)
        //      2、当前消费者的offset在kafka服务器上不存在(比如: 当前消费者消费的偏移量到了10，但是此时消费者宕机了很长一段时间，而服务器上数据默认保存7天，那么此时10之后的某些偏移量可能被删除了)
        // earliest: 从最早的偏移量开始消费
        // latest: 从最新的偏移量开始消费
        // none: 如果没有在消费者组中找到先前的偏移量，则向消费者抛出异常
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // 配置kafka自定义的拦截器,可以配置多个,多个以英文的逗号分割开
        properties.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, "com.huan.kafka.api.CustomConsumerInterceptor");
        // 设置两次 poll 方法之间的最大的延时，如果超过了最大的延时，则kafka认为该consumer消费能力弱，会将该分区给别的消费者消费
        properties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000);
        // 设置一次 pool 最多获取多少条记录
        properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        // 拉取消息时，每个分区返回的最大的消息的字节数
        properties.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG,1048576);

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        // 指定consumer消费的主题
        consumer.subscribe(Collections.singletonList("topic-a"));
        while (true) {
            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(500));
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println("接收到了一个数据. partition:" + consumerRecord.partition() + " offset:"
                        + consumerRecord.offset() + " 消息的值:" + consumerRecord.value());
                TimeUnit.SECONDS.sleep(3);
            }
        }
    }
}
